package me.riverslob.test.esper.plugin.esperio.rpc;

import com.espertech.esper.client.ConfigurationException;
import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EventType;
import com.espertech.esper.core.service.EPServiceProviderSPI;
import me.riverslob.test.esper.plugin.esperio.rpc.config.ConfigurationRPCAdapter;
import me.riverslob.test.esper.plugin.esperio.rpc.config.OutputConfig;
import me.riverslob.test.esper.plugin.esperio.rpc.input.EsperIORPCService;
import me.riverslob.test.esper.plugin.esperio.rpc.output.EsperIORPCSubscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * <br/>Created on 13-10-12 上午11:01
 *
 * @author 郑治明（zm.zheng@zhuche.com）
 * @since 3.2.2
 */
public class EsperIORPCAdapter {
    private static final Logger log = LoggerFactory.getLogger(EsperIORPCAdapter.class);

    private final ConfigurationRPCAdapter config;
    private final String engineURI;

    /**
     * Quickstart constructor.
     *
     * @param config configuration
     */
    public EsperIORPCAdapter(ConfigurationRPCAdapter config, String engineURI) {
        this.config = config;
        this.engineURI = engineURI;
    }

    /**
     * Re-initialize.
     */
    public void initialize() {
    }

    /**
     * Start the DDS endpoint.
     */
    public synchronized void start() {
        if (log.isInfoEnabled()) {
            log.info("Starting EsperIO HTTP Adapter for engine URI '" + engineURI + "'");
        }

        EPServiceProviderSPI engineSPI = (EPServiceProviderSPI) EPServiceProviderManager.getProvider(engineURI);

        // Start outputs (output adapter)
        for (OutputConfig output : config.getOutputs()) {
            EventType eventType = engineSPI.getEventAdapterService().getExistsTypeByName(output.getStream());
            if (eventType == null) {
                throw new ConfigurationException("Event type by name '" + output.getStream() + "' not found");
            }

            try {
                EsperIORPCSubscription subs = new EsperIORPCSubscription(output.getStream(), output.getProjects());
                subs.seteventTypeName(output.getStream());
                subs.setSubscriptionName("EsperIORPC-" + output.getStream() + "-" + output.getProjects());
                subs.registerAdapter(engineSPI);
            } catch (Throwable t) {
                log.error("Error starting RPC Output definition for stream " + output.getStream() + "'" + t.getMessage(), t);
            }
        }

        // Configure services (input adapter)
        if (config.getInput() != null && config.getInput().isOpen()) {
            EsperIORPCService inputService = new EsperIORPCService(engineSPI);
            inputService.start();
        }

        if (log.isInfoEnabled()) {
            log.info("Completed starting EsperIO HTTP Adapter for engine URI '" + engineURI + "'.");
        }
    }

    /**
     * Destroy the adapter.
     */
    public synchronized void destroy() {
        if (log.isDebugEnabled()) {
            log.debug("Destroying Esper RPC Adapter");
        }
    }

}
